package com.sankuai.inf.leaf.segment;

import com.sankuai.inf.leaf.IDGen;
import com.sankuai.inf.leaf.common.Result;
import com.sankuai.inf.leaf.common.Status;
import com.sankuai.inf.leaf.segment.dao.IDAllocDao;
import com.sankuai.inf.leaf.segment.model.*;
import org.perf4j.StopWatch;
import org.perf4j.slf4j.Slf4JStopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class SegmentIDGenImpl implements IDGen {
    private static final Logger logger = LoggerFactory.getLogger(SegmentIDGenImpl.class);

    /**
     * IDCache未初始化成功时的异常码
     */
    private static final long EXCEPTION_ID_IDCACHE_INIT_FALSE = -1;
    /**
     * key不存在时的异常码
     */
    private static final long EXCEPTION_ID_KEY_NOT_EXISTS = -2;
    /**
     * SegmentBuffer中的两个Segment均未从DB中装载时的异常码
     */
    private static final long EXCEPTION_ID_TWO_SEGMENTS_ARE_NULL = -3;
    /**
     * 最大步长不超过100,0000
     */
    private static final int MAX_STEP = 1000000;
    /**
     * 一个Segment维持时间为15分钟
     */
    private static final long SEGMENT_DURATION = 15 * 60 * 1000L;
    /**
     * 核心线程数5，最大线程数最大整数，空余线程存活60秒，一个长度等待队列的线程池
     */
    private ExecutorService service = new ThreadPoolExecutor(5, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new UpdateThreadFactory());
    /**
     * 第一次从数据库中获取记录更新cache是否完成
     */
    private volatile boolean initOK = false;
    /**
     * biz_tag->双buffer字典
     */
    private Map<String, SegmentBuffer> cache = new ConcurrentHashMap<String, SegmentBuffer>();
    private IDAllocDao dao;

    public static class UpdateThreadFactory implements ThreadFactory {

        private static int threadInitNumber = 0;

        private static synchronized int nextThreadNum() {
            return threadInitNumber++;
        }

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "Thread-Segment-Update-" + nextThreadNum());
        }
    }

    @Override
    public boolean init() {
        logger.info("Init ...");
        // 确保加载到kv后才初始化成功
        updateCacheFromDb();
        initOK = true;
        updateCacheFromDbAtEveryMinute();
        return initOK;
    }

    /**
     * 构造计划任务，每分钟调用一次updateCacheFromDb方法从数据库中更新cache
     */
    private void updateCacheFromDbAtEveryMinute() {
        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("check-idCache-thread");
                t.setDaemon(true);
                return t;
            }
        });
        service.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                updateCacheFromDb();
            }
        }, 60, 60, TimeUnit.SECONDS);
    }

    /**
     * 如果数据库中新增了tag，则构造双buffer结构到cache中
     * 如果数据库中移除了tag，则从cache中删除对应的数据
     */
    private void updateCacheFromDb() {
        logger.info("update cache from db");
        StopWatch sw = new Slf4JStopWatch();
        try {
            List<String> dbTags = dao.getAllTags();
            if (dbTags == null || dbTags.isEmpty()) {
                return;
            }
            // cache中所有biz_tag
            List<String> cacheTags = new ArrayList<String>(cache.keySet());
            // 当前数据库中所有biz_tag集合
            Set<String> insertTagsSet = new HashSet<>(dbTags);
            // 删除biz_tag集合=cache中所有biz_tag
            Set<String> removeTagsSet = new HashSet<>(cacheTags);
            //db中新加的tags灌进cache
            for(int i = 0; i < cacheTags.size(); i++){
                String tmp = cacheTags.get(i);
                // cache中所有biz_tag与当前数据库中所有biz_tag集合比较，过滤出本次数据库中新增的tags
                if(insertTagsSet.contains(tmp)){
                    insertTagsSet.remove(tmp);
                }
            }
            // 遍历新增的tag,构造双buffer结构
            for (String tag : insertTagsSet) {
                SegmentBuffer buffer = new SegmentBuffer();
                buffer.setKey(tag);
                Segment segment = buffer.getCurrent();
                segment.setValue(new AtomicLong(0));
                segment.setMax(0);
                segment.setStep(0);
                // 加入到cache中
                cache.put(tag, buffer);
                logger.info("Add tag {} from db to IdCache, SegmentBuffer {}", tag, buffer);
            }
            // cache中已失效的tags从cache删除，将数据库中所有tag和cache中所有tag比较，过滤出需要删除的tag
            // 具体逻辑为：如果数据库中不存在而cache中存在，那么这些tag就是要删除的
            for(int i = 0; i < dbTags.size(); i++){
                String tmp = dbTags.get(i);
                if(removeTagsSet.contains(tmp)){
                    removeTagsSet.remove(tmp);
                }
            }
            // 在cache中遍历删除需要移除的tag
            for (String tag : removeTagsSet) {
                cache.remove(tag);
                logger.info("Remove tag {} from IdCache", tag);
            }
        } catch (Exception e) {
            logger.warn("update cache from db exception", e);
        } finally {
            sw.stop("updateCacheFromDb");
        }
    }

    /**
     * 获取id
     *
     * @param key
     * @return
     */
    @Override
    public Result get(final String key) {
        // 如果还没有初始化完成，即从数据库中构造双buffer结构完成，抛出异常
        if (!initOK) {
            return new Result(EXCEPTION_ID_IDCACHE_INIT_FALSE, Status.EXCEPTION);
        }
        // 检查内存cache中是否包含这个key即biz_tag
        if (cache.containsKey(key)) {
            // 从cache中获取到这个key对应的双buffer
            SegmentBuffer buffer = cache.get(key);
            // 双buffer还未初始化完成，默认是未初始化完成
            if (!buffer.isInitOk()) {
                // 初始化时需要线程同步
                synchronized (buffer) {
                    // 双重检测
                    if (!buffer.isInitOk()) {
                        try {
                            updateSegmentFromDb(key, buffer.getCurrent());// 构造双buffer结构中当前正在使用的Segment
                            logger.info("Init buffer. Update leafkey {} {} from db", key, buffer.getCurrent());
                            // 设置双buffer已初始化完成
                            buffer.setInitOk(true);
                        } catch (Exception e) {
                            logger.warn("Init buffer {} exception", buffer.getCurrent(), e);
                        }
                    }
                }
            }
            // 双buffer已初始化完成，从双buffer中获取id
            return getIdFromSegmentBuffer(cache.get(key));
        }
        // 内存cache中不包含这个key时，抛出异常，初始化时和每分钟一次的周期任务会负责从数据库中更新cache
        return new Result(EXCEPTION_ID_KEY_NOT_EXISTS, Status.EXCEPTION);
    }

    /**
     * 从数据中中更新biz_tag对应的正在使用的Segment，一样是重中之重
     *
     * @param key biz_tag
     * @param segment 双buffer结构中当前正在使用的Segment
     */
    public void updateSegmentFromDb(String key, Segment segment) {
        StopWatch sw = new Slf4JStopWatch();
        // segment所属的双buffer
        SegmentBuffer buffer = segment.getBuffer();
        LeafAlloc leafAlloc;
        // 如果双buffer没有初始化完成
        if (!buffer.isInitOk()) {// 第一次初始化
            leafAlloc = dao.updateMaxIdAndGetLeafAlloc(key);
            // 设置双buffer的step和minstep属性值为查出记录的step值
            buffer.setStep(leafAlloc.getStep());
            buffer.setMinStep(leafAlloc.getStep());//leafAlloc中的step为DB中的step
        } else if (buffer.getUpdateTimestamp() == 0) {// 如果双buffer已初始化完成但更新时间戳为0,第二次,需要准备next Segment
            leafAlloc = dao.updateMaxIdAndGetLeafAlloc(key);
            // 设置更新时间戳
            buffer.setUpdateTimestamp(System.currentTimeMillis());
            // 设置双buffer的step和minstep属性值为查出记录的step值
            buffer.setStep(leafAlloc.getStep());
            buffer.setMinStep(leafAlloc.getStep());//leafAlloc中的step为DB中的step
        } else {
            // 三次及以上，动态设置nextStep（第一次是初始化第一个号段，第二次是准备下一个号段，第三次是第二个号段也满了，需要再一次准备下一个号段）
            // 时间差 = 当前时间戳 - 双buffer的更新时间戳
            long duration = System.currentTimeMillis() - buffer.getUpdateTimestamp();
            int nextStep = buffer.getStep();
            // 时间差 < 15分钟
            if (duration < SEGMENT_DURATION) {
                // 步长的2倍 > 一百万
                if (nextStep * 2 > MAX_STEP) {
                    //do nothing
                } else {
                    // 下一次步长 = 当前步长 * 2，
                    // 为了应对高并发下获取id的性能，将步长扩容后可以将max_id也更新为一个大的值，以承载更多请求
                    nextStep = nextStep * 2;
                }
            } else if (duration < SEGMENT_DURATION * 2) {// 15分钟 < 时间差 < 30分钟
                //do nothing with nextStep，此时不变化步长
            } else {// 超过了30分钟，说明并发量已经不大，此时缩容步长值，以减小max_id
                // 如果nextStep / 2后大于等于双buffer最小步长，则下一次步长设置为extStep / 2，否则设置为nextStep
                // 保证nextStep始终大于最小步长
                nextStep = nextStep / 2 >= buffer.getMinStep() ? nextStep / 2 : nextStep;
            }
            logger.info("leafKey[{}], step[{}], duration[{}mins], nextStep[{}]", key, buffer.getStep(), String.format("%.2f",((double)duration / (1000 * 60))), nextStep);
            LeafAlloc temp = new LeafAlloc();
            temp.setKey(key);
            temp.setStep(nextStep);
            // 入库
            leafAlloc = dao.updateMaxIdByCustomStepAndGetLeafAlloc(temp);
            // 设置双buffer的更新时间戳
            buffer.setUpdateTimestamp(System.currentTimeMillis());
            // 设置双buffer的step属性
            buffer.setStep(nextStep);
            // 设置双buffer的minstep属性，始终是数据库的step字段
            buffer.setMinStep(leafAlloc.getStep());//leafAlloc的step为DB中的step
        }
        // max_id - step
        long value = leafAlloc.getMaxId() - buffer.getStep();
        // 设置segment的当前下发值
        segment.getValue().set(value);
        // 设置segment的最大id
        segment.setMax(leafAlloc.getMaxId());
        // 设置segment的步长=双buffer的步长
        segment.setStep(buffer.getStep());
        sw.stop("updateSegmentFromDb", key + " " + segment);
    }

    /**
     * 从双buffer中获取唯一id，应该是重中之重的方法了吧
     *
     * @param buffer
     * @return
     */
    public Result getIdFromSegmentBuffer(final SegmentBuffer buffer) {
        while (true) {
            // 双buffer的读锁锁定，阻塞写行为
            buffer.rLock().lock();
            try {
                // 获取双buffer中当前正在使用的号段segment
                final Segment segment = buffer.getCurrent();
                // 如果下一个segment还不可切换 && 剩余id数量到达百分之90的阈值 && 双buffer的threadRunning属性可以由false改为true，就是说没有buffer线程在执行
                // 下面就开始准备第二个段（Segment）
                if (!buffer.isNextReady() && (segment.getIdle() < 0.9 * segment.getStep())
                        && buffer.getThreadRunning().compareAndSet(false, true)) {
                    service.execute(new Runnable() {
                        @Override
                        public void run() {
                            // 获取到当前未在使用的segment，即下一个segment
                            Segment next = buffer.getSegments()[buffer.nextPos()];
                            boolean updateOk = false;
                            try {
                                // 准备下一个号段时的更新segment
                                updateSegmentFromDb(buffer.getKey(), next);
                                // 更新segment完成
                                updateOk = true;
                                logger.info("update segment {} from db {}", buffer.getKey(), next);
                            } catch (Exception e) {
                                logger.warn(buffer.getKey() + " updateSegmentFromDb exception", e);
                            } finally {
                                if (updateOk) {
                                    // 加双buffer写锁
                                    buffer.wLock().lock();
                                    // 双buffer的下一段已准备好
                                    buffer.setNextReady(true);
                                    // 设置双buffer无线程在准备下一段
                                    buffer.getThreadRunning().set(false);
                                    // 释放写锁
                                    buffer.wLock().unlock();
                                } else {
                                    // updateSegmentFromDb出现异常时，设置无线程在准备下一个segment
                                    buffer.getThreadRunning().set(false);
                                }
                            }
                        }
                    });
                }
                // 离开准备下一个segment的方法
                // 获取然后递增当前号段的下发id值，使用CAS方式避免生成重复id
                long value = segment.getValue().getAndIncrement();
                // 校验准备下发的id值是否小于当前号段的最大id值，是的话则构造Result并返回
                if (value < segment.getMax()) {
                    return new Result(value, Status.SUCCESS);
                }
            } finally {
                // 释放双buffer的读锁
                buffer.rLock().unlock();
            }
            // 等待准备下一个号段的操作执行完成
            waitAndSleep(buffer);
            // 对双buffer加写锁
            buffer.wLock().lock();
            try {
                // 获取当前使用的号段
                final Segment segment = buffer.getCurrent();
                // 获取当前号段准备下发的id，并递增这个id
                // 为什么重复获取value, 多线程执行时,在进行waitAndSleep() 后,当前Segment可能已经被调换了，
                // 直接进行一次获取value的操作,可以提高id下发的速度(没必要再走一次循环)，
                // 并且防止出错(在交换Segment前进行一次检查)
                long value = segment.getValue().getAndIncrement();
                // 校验准备下发的id值是否小于当前号段的最大id值，是的话则构造Result并返回
                if (value < segment.getMax()) {
                    return new Result(value, Status.SUCCESS);
                }
                // 运行到这里说明当前号段的id已经下发完了
                if (buffer.isNextReady()) {
                    // 如果下一个号段已经准备好，则切换到下一个号段，并设置下一个号段准备好的标记为false
                    // 切换完成后回到while循环开始的地方重新获取id
                    buffer.switchPos();
                    buffer.setNextReady(false);
                } else {
                    // 如果下一个号段没有准备好，抛出异常
                    logger.error("Both two segments in {} are not ready!", buffer);
                    return new Result(EXCEPTION_ID_TWO_SEGMENTS_ARE_NULL, Status.EXCEPTION);
                }
            } finally {
                // 解除写锁
                buffer.wLock().unlock();
            }
        }
    }

    /**
     * 自旋等待
     *
     * @param buffer
     */
    private void waitAndSleep(SegmentBuffer buffer) {
        int roll = 0;// 自旋次数
        while (buffer.getThreadRunning().get()) {
            // 如果双buffer的准备下一个号段的操作有线程正在运行中，则自旋+1
            roll += 1;
            if(roll > 10000) {
                // 如果自旋超过了一万次则休眠10毫秒
                try {
                    TimeUnit.MILLISECONDS.sleep(10);
                    // 退出自旋过程
                    break;
                } catch (InterruptedException e) {
                    logger.warn("Thread {} Interrupted",Thread.currentThread().getName());
                    break;
                }
            }
        }
    }

    public List<LeafAlloc> getAllLeafAllocs() {
        return dao.getAllLeafAllocs();
    }

    public Map<String, SegmentBuffer> getCache() {
        return cache;
    }

    public IDAllocDao getDao() {
        return dao;
    }

    public void setDao(IDAllocDao dao) {
        this.dao = dao;
    }
}
